package cn.liwq.demo.disruptor;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.util.RandomUtil;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.stream.IntStream;
import java.util.stream.LongStream;

/**
 * @author lweiqiang
 * @date 2023/5/4
 */
@Slf4j
@Service
public class TaskProducer {

    @SneakyThrows
    public Disruptor<TaskEvent> send(int len) {
        int ringBufferSize = 1024;
        Disruptor<TaskEvent> disruptor = new Disruptor<>(TaskEvent::new, ringBufferSize,
                ThreadFactoryBuilder.create().setNamePrefix("disruptor-").build(), ProducerType.SINGLE,
                new BlockingWaitStrategy());
        TaskEventHandler taskEventHandler = new TaskEventHandler(disruptor);
        int consumers = 5;
        TaskEventHandler[] taskWorkers = new TaskEventHandler[consumers];
        IntStream.range(0, consumers).forEach(i -> taskWorkers[i] = taskEventHandler);
        EventHandlerGroup<TaskEvent> eventHandlerGroup = disruptor.handleEventsWithWorkerPool(taskWorkers)
                .then(new TaskResultEventHandler());
        SequenceBarrier sequenceBarrier = eventHandlerGroup.asSequenceBarrier();
        disruptor.setDefaultExceptionHandler(new TaskExceptionHandler(disruptor));
        disruptor.start();

        IntStream.range(0, len + 1).forEach(i -> {
            log.info("send {}", i);
            disruptor.publishEvent((e, seq) -> {
                e.setId(i);
                e.setContent(RandomUtil.randomString(8));
            });
        });
        long processedSeq = sequenceBarrier.waitFor(len);
        LongStream.range(0, processedSeq + 1).forEach(l -> {
            log.info("result-{}-{}", l, disruptor.getRingBuffer().get(l).getResult());
        });
        return disruptor;
    }
}
